package com.doit.demo.day02;


import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;


import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * @DATE 2022/2/15/15:15
 * @Author MDK
 * @Version 2021.2.2
 **/
public class MySqlFunction extends RichMapFunction<String, Tuple5<String, String, Integer, String, Double>> {
    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection("jdbc:mysql://linux01:3306/doit?characterEncoding=utf-8","root","root");
    }

    @Override
    public Tuple5<String, String, Integer, String, Double> map(String line) throws Exception {
        String[] fields = line.split(",");
        String uid = fields[0];
        String oid = fields[1];
        int cid = Integer.parseInt(fields[2]);
        double money = Double.parseDouble(fields[3]);

        PreparedStatement preparedStatement = connection.prepareStatement("select name from tb_category where id = ?");
        preparedStatement.setInt(1,cid);
        ResultSet resultSet = preparedStatement.executeQuery();


        String name = null;
        if(resultSet.next()){
            name = resultSet.getString(1);
        }else{
            name= "未知";
        }
        resultSet.close();
        preparedStatement.close();

        return Tuple5.of(uid,oid,cid,name,money);

    }

    @Override
    public void close() throws Exception {
        connection.close();
    }
}
